package org.example;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 *  MapReduce数据清洗
 */
public class MapReduceClean {

    private static String in = "/user/hive/warehouse/ods.db/book/*";
    private static String out = "/user/hive/warehouse/dws.db/book";

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        System.setProperty("HADOOP_USER_NAME","root");

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        job.setJarByClass(MapReduceClean.class);
        job.setMapperClass(BookMapper.class);

        // 设置输出类型
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);

        // 设置reduce个数为0
        job.setNumReduceTasks(0);

        // 设置输入和输出路径
        FileInputFormat.addInputPath(job, new Path(in));
        FileOutputFormat.setOutputPath(job, new Path(out));

        job.waitForCompletion(true);
    }


    public static class BookMapper extends Mapper<LongWritable, Text, NullWritable,Text>{

        @Override
//        key 行号 value内容
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, NullWritable, Text>.Context context) throws IOException, InterruptedException {

            String line = value.toString();
//            System.out.println(line);

            String[] fields = line.split("#");

            if(fields.length == 13){
                String category = fields[0];
                String home_url = fields[1];
                String img_url = fields[2];
                String name = fields[3];
                String comment = fields[4];
                String recommend = fields[5];
                String author = fields[6];
                String publication_time = fields[7];
                String publishing_house = fields[8];
                String price = fields[9];
                String ori_price = fields[10];
                String discount = fields[11];
                String descr = fields[12];


//                清洗评论值格式
                Integer commentNum = 0;
//                非空且有长度会进行解析
                if(StringUtils.isNotEmpty(comment) && !comment.isEmpty() && !"条评论".equals(comment)){
//                     过滤掉有脏数据的行
                    if(!comment.contains("条评论")){
                        return;
                    }
                    commentNum = Integer.valueOf(comment.replace("条评论",""));
                }

//                 * 清洗推荐值格式
                Double recommendNum = 0.0D;
                if(StringUtils.isNotEmpty(recommend) && !recommend.isEmpty()){
                    recommendNum = Double.valueOf(recommend.replace("%推荐",""));
                }

//                 清洗价格格式
                Double newPrice = 0.0D;
                if(StringUtils.isNotEmpty(price)){
                    newPrice = Double.valueOf(price.replace("¥","").replace(",",""));
                }

//                  清洗原价格式
                Double newOriPrice = 0.0D;
                if(StringUtils.isNotEmpty(ori_price)){
                    newOriPrice = Double.valueOf(ori_price.replace("¥","").replace(",",""));
                }

                String output = category + "#" + home_url + "#" + img_url + "#" + name + "#" + comment + "#" + recommend
                        + "#" + author + "#" + publication_time + "#" + publishing_house + "#" + newPrice + "#" + newOriPrice
                        + "#" +discount+ "#" +descr+ "#" +commentNum+ "#" +recommendNum;

                context.write(NullWritable.get(), new Text(output));
            }
        }
    }
}
